package com.tony.reactiveStream;

import org.reactivestreams.Processor;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.function.Function;

/**
 * @version V1.0  </br>
 * @ClassName MyProcessor </br>
 * @Description TODO  </br>
 * @Author tonywang</br>
 * @DATE 3/14/2019 5:24 PM</br>
 */
public class MyProcessor implements Processor<Employee, Freelancer> {
    private Subscription subscription;
    private Function<Employee,Freelancer> function;

    public MyProcessor(Function<Employee,Freelancer> function) {
        super();
        this.function = function;
    }
    @Override
    public void subscribe(Subscriber<? super Freelancer> s) {
        System.out.println("subscribe");
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(Employee pEmployee) {
        submit((Freelancer) function.apply(pEmployee));
        subscription.request(1);
    }

    private void  submit(Freelancer pFreelancer) {
        System.out.println("enter submit " + pFreelancer );
    }

    @Override
    public void onError(Throwable t) {
        t.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("Done");
    }

    public void close() {
        System.out.println("enter close, release resource");
    }
}

